package com.xiaoyu.im.concurrent;

import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import in.srain.cube.util.CLog;

public abstract class MessageQueueProcessor<T> implements Runnable {

    private final ExecutorService mExecutorService = new ThreadPoolExecutor(0, 1, 60L, TimeUnit.SECONDS, new LinkedBlockingQueue<>());
    private final ConcurrentLinkedQueue<T> mMessageQueue;
    private final String mName;

    public MessageQueueProcessor(String name) {
        mMessageQueue = new ConcurrentLinkedQueue<>();
        mName = name;
    }

    public void enqueueMessage(T message) {
        CLog.d(mName, "enqueueMessage: %s", message);
        mMessageQueue.add(message);
        mExecutorService.submit(this);
    }

    private void start() {
        CLog.d(mName, "start");
        onStart();

        while (!mMessageQueue.isEmpty()) {
            T message = mMessageQueue.poll();
            try {
                processMessage(message);
            } catch (Exception ex) {
                ex.printStackTrace();
                CLog.e(mName, "processMessage error", ex);
            }
            CLog.d(mName, "waiting for next message");
        }
    }

    private void finish() {
        CLog.d(mName, "all message has been processed, stop");
        onFinish();
    }

    protected abstract void processMessage(T message);

    protected void onStart() {
    }

    protected void onFinish() {
    }

    @Override
    public void run() {
        try {
            start();
        } catch (Exception ex) {
            CLog.e(mName, "run error", ex);
        } finally {
            finish();
        }
    }

    public void stop() {
        mMessageQueue.clear();
    }
}
